package ru.unlocker.reports.domain;

import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.stream.Stream;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import ru.unlocker.reports.domain.ReportStateMachine.ReportState;

public class StatusProcessor extends KeyedProcessFunction<String, ReportAndTimerEvent, ReportStatus> {

  private static final ReportStateMachine STATE_MACHINE = new ReportStateMachine();

  private ValueState<Long> bestBeforeTime;
  private ValueState<ReportState> previousState;

  @Override
  public void processElement(
      ReportAndTimerEvent value,
      KeyedProcessFunction<String, ReportAndTimerEvent, ReportStatus>.Context ctx,
      Collector<ReportStatus> out
  ) throws Exception {
    ReportState initialState = previousState.value();
    if (initialState == null) {
      ReportState newState = null;
      if (value.getReportEvent() == ReportEventKind.REPORT_STARTED) {
        newState = ReportState.STARTED_INACTIVE;
      } else if (value.getTimerEvent() == TimerEventKind.TIMER_ENABLED) {
        newState = ReportState.WAITING;
      }
      previousState.update(newState);
      Instant createdAt = value.getCreatedAt();
      long bestBeforeTimeMillis = createdAt.plus(1, ChronoUnit.DAYS).toEpochMilli();
      ctx.timerService().registerEventTimeTimer(bestBeforeTimeMillis);
      bestBeforeTime.update(bestBeforeTimeMillis);
      out.collect(new ReportStatus(value.getReportCode(), value.getWorkday(), newState, createdAt));
      return;
    }
    STATE_MACHINE.transfer(initialState, value).ifPresent(out::collect);
  }

  @Override
  public void open(OpenContext openContext) {
    ValueStateDescriptor<Long> bestBeforeTimeDescriptor = new ValueStateDescriptor<>("bestBeforeTime", Long.class);
    ValueStateDescriptor<ReportState> initialStateDescriptor = new ValueStateDescriptor<>(
        "previousState", ReportState.class
    );
    bestBeforeTime = getRuntimeContext().getState(bestBeforeTimeDescriptor);
    previousState = getRuntimeContext().getState(initialStateDescriptor);
  }

  @Override
  public void onTimer(
      long timestamp,
      KeyedProcessFunction<String, ReportAndTimerEvent, ReportStatus>.OnTimerContext ctx,
      Collector<ReportStatus> out
  ) {
    Stream.of(bestBeforeTime, previousState).forEach(ValueState::clear);
  }
}
